package com.baidu.cleaner;

import groovy.lang.GroovyClassLoader;
import groovy.lang.GroovyObject;

import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;

import com.baidu.automata.factory.CleanerFactory;
import com.baidu.cleaner.checker.Checker;
import com.baidu.cleaner.config.TD;
import com.baidu.cleaner.config.TR;
import com.baidu.cleaner.config.Table;
import com.baidu.cleaner.transformer.Transformer;

public abstract class BaseCleaner {
	private final static Logger logger = Logger.getLogger("flow_Platform");

	protected final static String lineSeparator = System.getProperty("line.separator");
	protected String inputFieldSeparator = null;
	protected String outputFieldSeparator = null;
	protected String cleanerId = null;
	protected long rowNum = 0;
	protected long failedNum = 0;
	protected Table table = null;
	protected TR tr = null;
	protected GroovyObject filter = null;
	protected GroovyObject selector = null;

	private final ReentrantLock readLock = new ReentrantLock();
	private final ReentrantLock writeLock = new ReentrantLock();
	private final ReentrantLock rowNumLock = new ReentrantLock();
	private final ReentrantLock failedNumLock = new ReentrantLock();
	private final ReentrantLock logLock = new ReentrantLock();

	public BaseCleaner(String inputFieldSeparator, String outputFieldSeparator, String cleanerId) throws Exception {
		super();
		this.inputFieldSeparator = inputFieldSeparator;
		this.outputFieldSeparator = outputFieldSeparator;
		this.cleanerId = cleanerId;
		getCleaner();
	}

	public void increaseRowNum() {
		rowNumLock.lock();
		this.rowNum++;
		rowNumLock.unlock();
	}

	public void increaseFailedNum() {
		failedNumLock.lock();
		this.failedNum++;
		failedNumLock.unlock();
	}

	public String getInputFieldSeparator() {
		return inputFieldSeparator;
	}

	public void setInputFieldSeparator(String inputFieldSeparator) {
		this.inputFieldSeparator = inputFieldSeparator;
	}

	public String getOutputFieldSeparator() {
		return outputFieldSeparator;
	}

	public void setOutputFieldSeparator(String outputFieldSeparator) {
		this.outputFieldSeparator = outputFieldSeparator;
	}

	public String getCleanerId() {
		return cleanerId;
	}

	public void setCleanerId(String cleanerId) {
		this.cleanerId = cleanerId;
	}

	public long getRowNum() {
		return rowNum;
	}

	public void setRowNum(long rowNum) {
		this.rowNum = rowNum;
	}

	public long getFailedNum() {
		return failedNum;
	}

	public void setFailedNum(long failedNum) {
		this.failedNum = failedNum;
	}

	public Table getTable() {
		return table;
	}

	public void setTable(Table table) {
		this.table = table;
	}

	protected boolean checkTD(String tdValue, Checker c) throws Exception {
		return c.verify(tdValue);
	}

	protected String transformTD(String tdValue, Transformer t) throws Exception {
		return t.transform(tdValue);
	}

	protected boolean parseTD(List<String> rowData, TD td, int index, long curRowNum) {
		String tdValue = null;
		if (td.getIndex() != -1) {
			tdValue = rowData.get(td.getIndex());
		}

		try {
			if (td.getChecker() != null && !checkTD(tdValue, td.getChecker())) {
				this.log(String.format("field[%d, %d] failed", curRowNum, td.getIndex() + 1));
				logger.error(String.format("field[%d, %d] failed", curRowNum, td.getIndex() + 1));
				return false;
			}

			if (td.getTransformer() != null) {
				tdValue = transformTD(tdValue, td.getTransformer());
				if (td.getIndex() == -1) {
					rowData.add(index, tdValue);
				} else {
					rowData.set(td.getIndex(), tdValue);
				}
			}
		} catch (Exception e) {
			this.log(String.format("field[%d, %d] failed, reason:%s", curRowNum, td.getIndex() + 1, e.getMessage()));
			logger.error(String.format("field[%d, %d] failed, reason:%s", curRowNum, td.getIndex() + 1, e.getMessage()));
			return false;
		}

		return true;
	}

	@SuppressWarnings("unchecked")
	protected void parseTR() {
		String trValue = null;
		long curRowNum = 0;
		try {
			while (true) {
				readLock.lock();
				try {
					trValue = this.getLine();
					if (trValue == null) {
						break;
					}
					this.rowNum++;
					curRowNum = this.rowNum;
				} catch (Exception e) {
					throw e;
				} finally {
					readLock.unlock();
				}

				trValue = StringUtils.isBlank(trValue) ? "" : trValue;

				List<String> fields = Collections.synchronizedList(Arrays.asList(trValue.split(this.inputFieldSeparator)));

				if (tr.getFieldNum() != fields.size()) {
					this.log(String.format("row[%d] failed, filed num is not right, actual is %d, expect is %d", curRowNum, fields.size(), tr.getFieldNum()));
					logger.error(String.format("row[%d] failed, filed num is not right, actual is %d, expect is %d", curRowNum, fields.size(), tr.getFieldNum()));
					this.increaseFailedNum();
					continue;
				}

				boolean hasError = false;
				List<TD> tds = tr.getTds();
				for (int i = 0; i < tds.size(); i++) {
					TD td = tds.get(i);
					if (!parseTD(fields, td, i, curRowNum)) {
						hasError = true;
					}
				}

				if (!hasError) {
					if (this.filter != null) {
						Boolean isOk = (Boolean) this.filter.invokeMethod("filter", fields.toArray());
						if (!isOk) {
							continue;
						}
					}

					if (this.selector != null) {
						List<String> rowList = (List<String>) this.selector.invokeMethod("select", fields.toArray());
						fields = rowList;
					}

					writeLock.lock();
					try {
						this.writeLine(StringUtils.join(fields, this.outputFieldSeparator));

					} catch (Exception e) {
						throw e;
					} finally {
						writeLock.unlock();
					}
				} else {
					this.increaseFailedNum();
				}
			}
		} catch (Exception e) {
			this.log(String.format("row[%d] failed, reason:%s", curRowNum, e.getMessage()));
			logger.error(String.format("row[%d] failed, reason:%s", curRowNum, e.getMessage()));
			this.increaseFailedNum();
		}
	}

	protected boolean parseTable(Table table) {
		tr = table.getTrs().get(0);
		try {
			if (!StringUtils.isBlank(tr.getFilterScript())) {
				ClassLoader cl = BaseCleaner.class.getClassLoader();
				GroovyClassLoader loader = new GroovyClassLoader(cl);
				Class<?> groovyClass = loader.parseClass(tr.getFilterScript());
				this.filter = (GroovyObject) groovyClass.newInstance();
			}

			if (!StringUtils.isBlank(tr.getSelectScript())) {
				ClassLoader cl = BaseCleaner.class.getClassLoader();
				GroovyClassLoader loader = new GroovyClassLoader(cl);
				Class<?> groovyClass = loader.parseClass(tr.getSelectScript());
				this.selector = (GroovyObject) groovyClass.newInstance();
			}
		} catch (Exception e) {
			this.log(e.getMessage());
			logger.error(e.getMessage());
			return false;
		}

		logger.info(String.format("start cleaning row data by %d threads", 5));

		Thread t1 = new Thread(new LineRunner(this));
		Thread t2 = new Thread(new LineRunner(this));
		Thread t3 = new Thread(new LineRunner(this));
		Thread t4 = new Thread(new LineRunner(this));
		Thread t5 = new Thread(new LineRunner(this));
		t1.start();
		t2.start();
		t3.start();
		t4.start();
		t5.start();

		try {
			t1.join();
			t2.join();
			t3.join();
			t4.join();
			t5.join();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

		return this.failedNum > 0 ? false : true;
	}

	public void getCleaner() throws Exception {
		this.table = CleanerFactory.getCleaner(this.cleanerId);
	}

	protected abstract String getLine() throws Exception;

	protected abstract void writeLine(String line) throws Exception;

	protected abstract void writeLog(String msg) throws Exception;

	protected void log(String msg) {
		logLock.lock();
		try {
			writeLog(msg);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			logLock.unlock();
		}
	}

	protected abstract void close();

	public final void parse() {
		parseTable(this.table);
		close();
	}
}
